package bigdata;

import java.util.Random;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hazelcast.config.Config;
import com.hazelcast.config.MapConfig;
import com.hazelcast.config.MapIndexConfig;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IMap;
import com.hazelcast.core.PartitionService;
import com.hazelcast.instance.GroupProperty;
import com.hazelcast.monitor.LocalMapStats;

public class PrimeMember {
	private final static Logger logger = LoggerFactory.getLogger(PrimeMember.class);

	public static void main(String[] args) throws InterruptedException {
		HazelcastInstance hz = Hazelcast.newHazelcastInstance(config());

		IMap<Integer, Rule> ruleMap = hz.getMap("ruleMap");

		// TODO generate rule map data ; more than 100,000 entries
		generateRuleMapData(ruleMap);
		logger.info("generate rule finised!");

		// health check
		PartitionService partitionService = hz.getPartitionService();
		while (true) {
			LocalMapStats mapStatistics = ruleMap.getLocalMapStats();
			logger.info(
					"isClusterSafe:{},isLocalMemberSafe:{},rulemap.size:{},backpressure.enabled:{}, \n number of entries owned on this node = {}",
					partitionService.isClusterSafe(), partitionService.isLocalMemberSafe(), ruleMap.size(),
					hz.getConfig().getProperty(GroupProperty.BACKPRESSURE_ENABLED), mapStatistics.getOwnedEntryCount());
			Thread.sleep(1000);
		}

	}

	public static Config config() {
		Config config = new Config();
		config.setProperty(GroupProperty.ENABLE_JMX, "true");
		config.setProperty(GroupProperty.BACKPRESSURE_ENABLED, "true");
		config.setProperty(GroupProperty.SLOW_OPERATION_DETECTOR_ENABLED, "true");

		config.getSerializationConfig().addPortableFactory(1, new MyPortableFactory());

		MapConfig mapConfig = new MapConfig("ruleMap");
		// map backup
		mapConfig.setBackupCount(0);
		mapConfig.setAsyncBackupCount(1);
		mapConfig.setReadBackupData(true);
		// rule map index
		mapConfig.addMapIndexConfig(new MapIndexConfig("tel", true));
		mapConfig.addMapIndexConfig(new MapIndexConfig("isBlack", false));
		mapConfig.addMapIndexConfig(new MapIndexConfig("isWhite", false));
		config.addMapConfig(mapConfig);
		
		config.getExecutorConfig("hz:query").setPoolSize(100);
		return config;
	}

	private static void generateRuleMapData(IMap<Integer, Rule> ruleMap) {
		Random random = new Random(System.currentTimeMillis());
		for (int i = 0; i < 1000000; i++) {
			int tel = random.nextInt() % 100000000;
			boolean isBlack = random.nextBoolean();
			boolean isWhite = random.nextBoolean();
			Rule rule = new Rule(tel, isBlack, isWhite);
			ruleMap.put(Integer.valueOf(tel), rule);

			if (i % 100000 == 0) {
				logger.info("i:{}", i);
			}
		}
	}
}
